﻿using System.Collections.Concurrent;
using System.Net.WebSockets;

namespace sgf.server
{
    public class WebSocketChannel : NetChannel
    {
        private WebSocket _webSocket;

        private readonly Action<long, byte[]> _onMessage;

        private readonly ConcurrentQueue<byte[]> _sendQueue = new();

        private readonly SemaphoreSlim _newSendMsgSemaphore = new(0);

        public WebSocketChannel(WebSocket webSocket, string remoteAddress, Action<long, byte[]> onMessage)
        {
            base.remoteAddress = remoteAddress;
            _webSocket = webSocket;
            _onMessage = onMessage;
        }

        protected override async void Close()
        {
            try
            {
                base.Close();
                await _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "socketclose", CancellationToken.None);
            }
            catch
            {
                // ignored
            }
            finally
            {
                _webSocket = null;
            }
        }

        public override async Task StartAsync()
        {
            try
            {
                _ = DoSend();
                await DoReceive();
            }
            catch (OperationCanceledException)
            {
            }
            catch (Exception e)
            {
                LogMgr.Error(e.Message);
            }
        }

        private async Task DoSend()
        {
            try
            {
                var array = new object[2];
                var closeToken = CloseSrc.Token;
                while (!closeToken.IsCancellationRequested)
                {
                    await _newSendMsgSemaphore.WaitAsync(closeToken);

                    if (!_sendQueue.TryDequeue(out var message))
                    {
                        continue;
                    }

                    await _webSocket.SendAsync(message, WebSocketMessageType.Binary, true, closeToken);
                }
            }
            catch
            {
                // ignored
            }
        }


        private async Task DoReceive()
        {
            var stream = new MemoryStream();
            var buffer = new ArraySegment<byte>(new byte[2048]);

            var closeToken = CloseSrc.Token;
            while (!closeToken.IsCancellationRequested)
            {
                WebSocketReceiveResult result;
                stream.SetLength(0);
                stream.Seek(0, SeekOrigin.Begin);
                do
                {
                    result = await _webSocket.ReceiveAsync(buffer, closeToken);
                    if (buffer.Array != null) stream.Write(buffer.Array, buffer.Offset, result.Count);
                } while (!result.EndOfMessage);

                if (result.MessageType == WebSocketMessageType.Close)
                    break;

                stream.Seek(0, SeekOrigin.Begin);
                _onMessage(GetData<long>("sessionId"), stream.GetBuffer());
            }

            stream.Close();
        }

        public override void Write(byte[] msg)
        {
            _sendQueue.Enqueue(msg);
            _newSendMsgSemaphore.Release();
        }
    }
}